package com.shujia.flink.window

import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object Demo4WindowCore {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)


    /**
      * 每隔5秒统计每个班级的平均年龄
      *
      */

    val kvDS: DataStream[(String, Double)] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      (split(4), split(2).toDouble)
    })


    val avgAgeDS: DataStream[(String, Double)] = kvDS
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))
      //flink 底层api 可以直接操作时间，事件和状态
      .process(new MyProcessWindowFunction)

    avgAgeDS.print()

    env.execute()

  }

}


class MyProcessWindowFunction extends ProcessWindowFunction[(String, Double), (String, Double), String, TimeWindow] {
  /**
    * 每一个key每一个窗口一次
    *
    */
  override def process(clazz: String, //key； 班级
                       context: Context, //上下文对象
                       elements: Iterable[(String, Double)], //一个key在一个窗口中所有的数据
                       out: Collector[(String, Double)] //用于将数据发送到下游
                      ): Unit = {


    //拿到一个班级在一个窗口内所有的年龄
    val ages: List[Double] = elements.map(_._2).toList

    //计算平均值
    val avgAge: Double = ages.sum / ages.length


    //将数据发送到下游
    out.collect((clazz, avgAge))
  }
}